package mylab.utils.common.concurrent;

import com.github.rholder.retry.RetryException;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableScheduledFuture;
import com.google.common.util.concurrent.SettableFuture;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import lombok.experimental.UtilityClass;

@UtilityClass
public class CallableUtil implements Concurrent {

    /**
     * 执行异步任务并等待执行结果返回
     *
     * @param callable 可执行任务
     */
    public <T> T runSync(Callable<T> callable) throws ExecutionException, InterruptedException {
        SettableFuture<T> future = createSettableFuture(callable);
        return future.get();
    }

    /**
     * 执行异步任务并等待执行结果返回
     *
     * @param callable 可执行任务
     * @param timeout  超时时间
     * @param unit     时间单位
     */
    public <T> T runSync(Callable<T> callable, long timeout, TimeUnit unit) throws ExecutionException, InterruptedException, TimeoutException {
        SettableFuture<T> future = createSettableFuture(callable);
        return future.get(timeout, unit);
    }

    public <T> T runSync(Callable<T> callable, int tryTimes) throws ExecutionException, RetryException {
        Preconditions.checkState(tryTimes > 0);
        Retryer<T> retryer = RetryerBuilder.<T>newBuilder()
                .retryIfException()
                .withWaitStrategy(WaitStrategies.noWait())
                .withStopStrategy(StopStrategies.stopAfterAttempt(tryTimes))
                .build();
        return runSync(callable, retryer);
    }

    /**
     * 执行异步任务并等待执行结果返回（含重试机制）
     *
     * @param callable 可执行任务
     * @param retryer  重试机制
     */
    public <T> T runSync(Callable<T> callable, Retryer<T> retryer) throws ExecutionException, RetryException {
        return retryer.call(callable);
    }

    /**
     * 执行异步任务，任务有返回值
     *
     * @param callable 可执行任务
     */
    public <T> ListenableFuture<T> submit(Callable<T> callable) {
        Preconditions.checkNotNull(callable);
        return listeningExecutor.submit(callable);
    }

    /**
     * 执行异步任务，任务有返回值
     *
     * @param callable 可执行任务
     * @param callback 异步回调
     */
    public <T> ListenableFuture<T> submit(Callable<T> callable, FutureCallback<T> callback) {
        Preconditions.checkNotNull(callable);
        Preconditions.checkNotNull(callback);
        ListenableFuture<T> future = listeningExecutor.submit(callable);
        Futures.addCallback(future, callback, listeningExecutor);
        return future;
    }

    public <T> ListenableFuture<T> submit(Callable<T> callable, FutureCallback<T> callback, int tryTimes) {
        Preconditions.checkState(tryTimes > 0);
        Retryer<T> retryer = RetryerBuilder.<T>newBuilder()
                .retryIfException()
                .withWaitStrategy(WaitStrategies.noWait())
                .withStopStrategy(StopStrategies.stopAfterAttempt(tryTimes))
                .build();
        return submit(callable, callback, retryer);
    }

    public <T> ListenableFuture<T> submit(Callable<T> callable, FutureCallback<T> callback, long timeout, TimeUnit unit) {
        Preconditions.checkNotNull(unit);
        Preconditions.checkState(timeout > 0);
        Retryer<T> retryer = RetryerBuilder.<T>newBuilder()
                .retryIfException()
                .withWaitStrategy(WaitStrategies.noWait())
                .withStopStrategy(StopStrategies.stopAfterDelay(timeout, unit))
                .build();

        return submit(callable, callback, retryer);
    }


    /**
     * 执行异步任务（含重试机制）
     *
     * @param callable 可执行任务
     * @param callback 回调方法
     * @param retryer  重试机制
     */
    public <T> ListenableFuture<T> submit(Callable<T> callable, FutureCallback<T> callback, Retryer<T> retryer) {
        Preconditions.checkNotNull(callable);
        Preconditions.checkNotNull(callback);
        ListenableFuture<T> future = submit(() -> retryer.call(callable), callback);
        Futures.addCallback(future, callback, listeningExecutor);
        return future;
    }


    /**
     * 延时执行
     *
     * @param callable 可执行任务，有返回值
     * @param delay    延时
     * @param unit     时间单位
     */
    public <T> ListenableScheduledFuture<T> schedule(Callable<T> callable, long delay, TimeUnit unit) {
        Preconditions.checkNotNull(callable);
        Preconditions.checkNotNull(unit);
        return listeningScheduledExecutor.schedule(callable, delay, unit);
    }

    private <T> SettableFuture<T> createSettableFuture(Callable<T> callable) {
        Preconditions.checkNotNull(callable);
        SettableFuture<T> settableFuture = SettableFuture.create();
        submit(callable, new FutureCallback<>() {
            @Override
            public void onSuccess(T result) {
                settableFuture.set(result);
            }

            @Override
            public void onFailure(Throwable t) {
                settableFuture.setException(t);
            }
        });

        return settableFuture;
    }
}
